package com.pw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.pw.gmall.realtime.app.BaseAppV2;
import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.entities.OrderWide;
import com.pw.gmall.realtime.entities.PaymentWide;
import com.pw.gmall.realtime.entities.ProductStats;
import com.pw.gmall.realtime.function.DimAsyncFunction;
import com.pw.gmall.realtime.utils.AtguiguUtils;
import com.pw.gmall.realtime.utils.DimUtils;
import com.pw.gmall.realtime.utils.FlinkSinkUtils;
import com.pw.gmall.realtime.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @Author: linux_future
 * DwdLogApp->DwdDbApp->DwmOrderWideAppV3->DwmPaymentWideApp
 * @since: 2022/3/25
 **/
public class ProductStatsApp extends BaseAppV2 {
    public static void main(String[] args) {
        new ProductStatsApp().init(2022, 1, "ProductStatsApp",
                "ProductStatsApp",
                Constant.TOPIC_DWD_PAGE_LOG,
                Constant.TOPIC_DWD_DISPLAY_LOG,
                Constant.TOPIC_DWM_ORDER_WIDE,
                Constant.TOPIC_DWD_FAVOR_INFO,
                Constant.TOPIC_DWD_CART_INFO,
                Constant.TOPIC_DWM_PAYMENT_WIDE,
                Constant.TOPIC_DWD_ORDER_REFUND_INFO,
                Constant.TOPIC_DWD_COMMENT_INFO
        );


    }

    @Override
    protected void handler(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> stream) {
        //解析多个流，合并成一个流
        DataStream<ProductStats> unionStream = etlStream(stream);
        //unionStream.print("union");
        //开窗聚合
        SingleOutputStreamOperator<ProductStats> aggStream = widowAndAgg(unionStream);
        //join维度信息  产品相关的维度信息: sku spu tm c3
        SingleOutputStreamOperator<ProductStats> joinStream = joinDIM(aggStream);
        //写入clickhouse
        writeToClickhouseSink(joinStream);
        //写入kafka数据,
        writeToKafkaSink(joinStream);


    }

    private void writeToKafkaSink(SingleOutputStreamOperator<ProductStats> stream) {
        stream.map(JSON::toJSONString)
                .addSink(KafkaUtils.getKafkaSink(Constant.TOPIC_DWS_PRODUCT_WIDE));
    }

    private void writeToClickhouseSink(SingleOutputStreamOperator<ProductStats> stream) {
        stream.addSink(FlinkSinkUtils.getClickhouseSink(Constant.CLICKHOUSE_DB, Constant.CLICKHOUSE_TABLE_PRODUCT_STATS, ProductStats.class));
    }

    private SingleOutputStreamOperator<ProductStats> joinDIM(SingleOutputStreamOperator<ProductStats> stream) {
        SingleOutputStreamOperator<ProductStats> joinStream = AsyncDataStream.unorderedWait(stream, new DimAsyncFunction<ProductStats>() {
            @Override
            public void addDim(Jedis jedis, Connection conn, ProductStats input, ResultFuture<ProductStats> resultFuture) throws Exception {
                // sku spu tm c3
                // ID  | SPU_ID  | PRICE  | SKU_NAME  | SKU_DESC  | WEIGHT  | TM_ID  | CATEGORY3_ID  | SKU_DEFAULT_IMG  | IS_SALE  | CREATE_TIME
                JSONObject skuInfo = DimUtils.readDim(jedis, conn, "dim_sku_info", input.getSku_id());
                input.setSku_name(skuInfo.getString("SKU_NAME"));
                input.setSku_price(skuInfo.getBigDecimal("PRICE"));

                input.setSpu_id(skuInfo.getLong("SPU_ID"));
                input.setTm_id(skuInfo.getLong("TM_ID"));
                input.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));

                //spu  ID  | SPU_NAME  | DESCRIPTION  | CATEGORY3_ID  | TM_ID
                JSONObject spu = DimUtils.readDim(jedis, conn, "dim_spu_info", input.getSpu_id());
                input.setSpu_name(spu.getString("SPU_NAME"));
                //dim_base_trademark
                JSONObject trademark = DimUtils.readDim(jedis, conn, "dim_base_trademark", input.getTm_id());
                input.setTm_name(trademark.getString("TM_NAME"));
                //c3
                JSONObject category3 = DimUtils.readDim(jedis, conn, "dim_base_category3", input.getCategory3_id());
                input.setCategory3_name(category3.getString("NAME"));
                resultFuture.complete(Collections.singletonList(input));
            }
        }, 60, TimeUnit.SECONDS);
        return joinStream;
    }

    private SingleOutputStreamOperator<ProductStats> widowAndAgg(DataStream<ProductStats> stream) {
        WatermarkStrategy<ProductStats> wms = WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((ps, ts) -> ps.getTs());
        SingleOutputStreamOperator<ProductStats> reduce = stream.assignTimestampsAndWatermarks(wms)
                .keyBy(ProductStats::getSku_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats ps1, ProductStats ps2) throws Exception {
                        ps1.setClick_ct(ps1.getClick_ct() + ps2.getClick_ct());
                        ps1.setDisplay_ct(ps1.getDisplay_ct() + ps2.getDisplay_ct());

                        ps1.setFavor_ct(ps1.getFavor_ct() + ps2.getFavor_ct());
                        ps1.setCart_ct(ps1.getCart_ct() + ps2.getCart_ct());

                        ps1.setOrder_sku_num(ps1.getOrder_sku_num() + ps2.getOrder_sku_num());
                        ps1.setOrder_amount(ps1.getOrder_amount().add(ps2.getOrder_amount()));
                        ps1.getOrderIdSet().addAll(ps2.getOrderIdSet());

                        ps1.setPayment_amount(ps1.getPayment_amount().add(ps2.getPayment_amount()));
                        ps1.getPaidOrderIdSet().addAll(ps2.getPaidOrderIdSet());

                        ps1.setRefund_amount(ps1.getRefund_amount().add(ps2.getRefund_amount()));
                        ps1.getRefundOrderIdSet().addAll(ps2.getRefundOrderIdSet());

                        ps1.setComment_ct(ps1.getComment_ct() + ps2.getComment_ct());
                        ps1.setGood_comment_ct(ps1.getGood_comment_ct() + ps2.getGood_comment_ct());


                        return ps1;
                    }
                }, new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                    @Override
                    public void process(Long aLong, Context ctx, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
                        ProductStats ps = elements.iterator().next();

                        ps.setStt(AtguiguUtils.toDateTime(ctx.window().getStart()));
                        ps.setEdt(AtguiguUtils.toDateTime(ctx.window().getEnd()));

                        ps.setTs(System.currentTimeMillis());

                        ps.setOrder_ct((long) ps.getOrderIdSet().size());
                        ps.setPaid_order_ct((long) ps.getPaidOrderIdSet().size());
                        ps.setRefund_order_ct((long) ps.getRefundOrderIdSet().size());

                        out.collect(ps);
                    }
                });
        return reduce;

    }

    private DataStream<ProductStats> etlStream(HashMap<String, DataStreamSource<String>> stream) {


        //click 点击
        SingleOutputStreamOperator<ProductStats> clickStream = stream.get(Constant.TOPIC_DWD_PAGE_LOG).flatMap(new FlatMapFunction<String, ProductStats>() {
            @Override
            public void flatMap(String value, Collector<ProductStats> out) throws Exception {
                JSONObject obj = JSON.parseObject(value);
                JSONObject page = obj.getJSONObject("page");
                String item_type = page.getString("item_type");
                //page 中sku_id存在，这表示点击
                if ("sku_id".equals(item_type)) {
                    ProductStats product = new ProductStats();
                    Long ts = obj.getLong("ts");
                    Long sku_id = page.getLong("item");
                    product.setSku_id(sku_id);
                    product.setClick_ct(1L);
                    product.setTs(ts);
                    out.collect(product);
                }
            }
        });
        //clickStream.print("click");
        //曝光
        SingleOutputStreamOperator<ProductStats> displayStream = stream.get(Constant.TOPIC_DWD_DISPLAY_LOG)
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value,
                                               Context ctx,
                                               Collector<ProductStats> out) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        // 判断是否为点击
                        String itemType = obj.getString("item_type");
                        if ("sku_id".equals(itemType)) {  //  表示点击了一个商品
                            ProductStats ps = new ProductStats();
                            Long ts = obj.getLong("ts");
                            Long sku_id = obj.getLong("item");
                            ps.setSku_id(sku_id);
                            ps.setDisplay_ct(1L);
                            ps.setTs(ts);
                            out.collect(ps);
                        }
                    }
                });
        // displayStream.print("display");
        //收藏
        SingleOutputStreamOperator<ProductStats> favorStream = stream.get(Constant.TOPIC_DWD_FAVOR_INFO)
                .map(obj -> {
                            JSONObject favorInfo = JSON.parseObject(obj);
                            long ts = AtguiguUtils.dateTimeToTs(favorInfo.getString("create_time"));
                            Long sku_id = favorInfo.getLong("sku_id");
                            ProductStats ps = new ProductStats();
                            ps.setTs(ts);
                            ps.setSku_id(sku_id);
                            ps.setFavor_ct(1L);
                            return ps;
                        }
                );
        //favorStream.print("favor");

        //购物车
        SingleOutputStreamOperator<ProductStats> cartStream = stream.get(Constant.TOPIC_DWD_CART_INFO).map(obj -> {
            JSONObject cart = JSON.parseObject(obj);
            long ts = AtguiguUtils.dateTimeToTs(cart.getString("create_time"));
            Long sku_id = cart.getLong("sku_id");
            ProductStats ps = new ProductStats();
            ps.setTs(ts);
            ps.setSku_id(sku_id);
            ps.setCart_ct(1L);
            return ps;
        });
        //cartStream.print("cart");
        //退款
        SingleOutputStreamOperator<ProductStats> refundStream = stream.get(Constant.TOPIC_DWD_ORDER_REFUND_INFO)
                .map(obj -> {
                    JSONObject refund = JSON.parseObject(obj);
                    ProductStats ps = new ProductStats();
                    long ts = AtguiguUtils.dateTimeToTs(refund.getString("create_time"));
                    Long sku_id = refund.getLong("sku_id");
                    BigDecimal refund_amount = refund.getBigDecimal("refund_amount");
                    Long order_id = refund.getLong("order_id");
                    Set<Long> singleton = Collections.singleton(order_id);
                    ps.setRefund_amount(refund_amount);
                    ps.setTs(ts);
                    ps.setSku_id(sku_id);
                    ps.getRefundOrderIdSet().addAll(singleton);
                    return ps;
                });

        //转换评价流数据
        SingleOutputStreamOperator<ProductStats> commentStream = stream.get(Constant.TOPIC_DWD_COMMENT_INFO)
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<ProductStats> out) throws Exception {
                        JSONObject comment = JSON.parseObject(value);
                        long ts = AtguiguUtils.dateTimeToTs(comment.getString("create_time"));
                        Long sku_id = comment.getLong("sku_id");
                        ProductStats ps = new ProductStats();
                        ps.setSku_id(sku_id);
                        ps.setTs(ts);
                        ps.setComment_ct(1L);
                        Long appraise = comment.getLong("appraise");
                        long isGood = 0L;
                        if (1204 <= appraise) {
                            isGood = 1L;
                        }
                        ps.setGood_comment_ct(isGood);
                        out.collect(ps);
                    }
                });
        //commentStream.print("comment");
        //订单宽表
        SingleOutputStreamOperator<ProductStats> orderWideStream = stream.get(Constant.TOPIC_DWM_ORDER_WIDE).map(
                obj -> {
                    OrderWide orderWide = JSON.parseObject(obj, OrderWide.class);
                    long ts = AtguiguUtils.dateTimeToTs(orderWide.getCreate_time());
                    ProductStats ps = new ProductStats();
                    ps.setTs(ts);
                    ps.setSku_id(orderWide.getSku_id());
                    ps.setSku_name(orderWide.getSku_name());
                    ps.getOrderIdSet().addAll(Collections.singleton(orderWide.getOrder_id()));
                    ps.setOrder_amount(orderWide.getSplit_total_amount());
                    return ps;
                }
        );
        SingleOutputStreamOperator<ProductStats> payStream = stream.get(Constant.TOPIC_DWM_PAYMENT_WIDE).map(obj -> {
            PaymentWide pay = JSON.parseObject(obj, PaymentWide.class);
            long ts = AtguiguUtils.dateTimeToTs(pay.getPayment_create_time());
            ProductStats ps = new ProductStats();
            ps.setTs(ts);
            ps.setSku_id(pay.getSku_id());
            ps.setPayment_amount(pay.getSplit_total_amount());
            ps.getPaidOrderIdSet().addAll(Collections.singleton(pay.getOrder_id()));
            return ps;
        });
        DataStream<ProductStats> unionStream = orderWideStream.union(payStream, commentStream, cartStream, favorStream, displayStream, clickStream, refundStream);
        return unionStream;
    }
}
